package com.yz.kafka.producers;

import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * @author yuanzheng
 * @since 2021/4/7
 */
public class CustomStudentSerializer implements Serializer<Student> {

    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public byte[] serialize(String s, Student student) {
        if (student == null) {
            return null;
        }
        byte[] name, address;
        if (student.getName() != null) {
            name = student.getName().getBytes(StandardCharsets.UTF_8);
        } else {
            name = new byte[0];
        }
        if (student.getAddress() != null) {
            address = student.getAddress().getBytes(StandardCharsets.UTF_8);
        } else {
            address = new byte[0];
        }
        ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + name.length + address.length);
        buffer.putInt(name.length);
        buffer.put(name);
        buffer.putInt(address.length);
        buffer.put(address);
        return buffer.array();
    }

    @Override
    public void close() {

    }
}
